package com.源码分析.using;

import java.util.Random;

import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

// 与using操作符类似，usingWhen使我们能以响应式方式管理资源。但是using会同步获取资源。usingWhen是响应式的获取资源。
public class UsingWhenDemo {

	private static final Logger logger = LoggerFactory.getLogger(UsingWhenDemo.class);

	public static void main(String[] args) throws InterruptedException {
		Flux.usingWhen(
				Transaction.beginTransaction(), // 通过返回Mono<Transaction>异步返回一个新事务
				transaction -> transaction.insertRows(Flux.just("A", "B", "C")), // 对于给定的事务实例，他会尝试插入新行
				transaction -> transaction.commit(), // 如果步骤2成功完成，则提交事务
				transaction -> transaction.rollback() // 如果步骤2失败，则回滚事务
		)
		.subscribe(
				d -> logger.info("onNext: {}", d),
				e -> logger.info("onError: {}", e.getMessage()),
				() -> logger.info("onComplete")
		);
		Thread.sleep(5000000);
	}

	private static class Transaction {
		private static final Random random = new Random();
		private final int id;

		public Transaction(int id) {
			this.id = id;
			logger.info("[T: {}] created", id);
		}

		public static Mono<Transaction> beginTransaction(){ // 静态工厂，能够创建新事物
			return Mono.defer(() ->
				Mono.just(new Transaction(random.nextInt(1000))));
		}

		public Flux<String> insertRows(Publisher<String> rows){ // 每个事务都有一种在事务中保存新行的方法，有时，某些内部问题会导致该过程失败，insertRows会消费并返回响应式流
			return Flux.from(rows)
					//.delayElements(Duration.ofMillis(100)) // 可以实现异步延迟
					.flatMap(r -> {
						if(random.nextInt(10) < 2) {
							return Mono.error(new RuntimeException("error: " + r));
						}else {
							return Mono.just(r);
						}
					});
		}

		public Mono<Void> commit(){ // 这是一个异步提交，有时，事务可能无法提交
			return Mono.defer(() -> {
				logger.info("T[: {}] commit", id);
				if(random.nextBoolean()) {
					return Mono.empty();
				}else {
					return Mono.error(new RuntimeException("conflict"));
				}
			});
		}

		public Mono<Void> rollback(){ // 这是一个异步回滚，有时，事务可能无法回滚
			return Mono.defer(() -> {
				logger.info("T[: {}] rollback", id);
				if(random.nextBoolean()) {
					return Mono.empty();
				}else {
					return Mono.error(new RuntimeException("conn error"));
				}
			});
		}
	}

}
